package com.example.mgsnew.config;


import com.fasterxml.jackson.annotation.JsonTypeInfo;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.ConditionalRejectingErrorHandler;
import org.springframework.amqp.rabbit.listener.FatalExceptionStrategy;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConversionException;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.MessageDeliveryException;
import org.springframework.retry.backoff.ExponentialBackOffPolicy;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;

@Slf4j
@Configuration
public class RabbitMQConfig {

    @Value("${spring.rabbitmq.host:127.0.0.1}")
    private String host;

    @Value("${spring.rabbitmq.port:5672}")
    private int port;

    @Value("${spring.rabbitmq.username:guest}")
    private String username;

    @Value("${spring.rabbitmq.password:guest}")
    private String password;

    @Value("${spring.rabbitmq.virtual-host:/}")
    private String virtualHost;

    @Value("${spring.rabbitmq.listener.simple.concurrency:5}")
    private int concurrency;

    @Value("${spring.rabbitmq.listener.simple.max-concurrency:10}")
    private int maxConcurrency;

    @Value("${spring.rabbitmq.listener.simple.retry.initial-interval:1000}")
    private long initialInterval;

    @Value("${spring.rabbitmq.listener.simple.retry.max-attempts:3}")
    private int maxAttempts;

    @Value("${spring.rabbitmq.listener.simple.retry.max-interval:10000}")
    private long maxInterval;

    @Value("${spring.rabbitmq.listener.simple.retry.multiplier:2.0}")
    private double multiplier;

    @Bean
    public CachingConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setHost(host);
        connectionFactory.setPort(port);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        connectionFactory.setVirtualHost(virtualHost);
        // 设置发布确认
        connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
        // 设置发布返回
        connectionFactory.setPublisherReturns(true);
        // 设置连接池大小
        connectionFactory.setChannelCacheSize(25);
        return connectionFactory;
    }

    @Bean
    public MessageConverter jsonMessageConverter() {
        ObjectMapper objectMapper = new ObjectMapper();
        // 配置类型信息处理
        objectMapper.activateDefaultTyping(
                objectMapper.getPolymorphicTypeValidator(),
                ObjectMapper.DefaultTyping.NON_FINAL,
                JsonTypeInfo.As.PROPERTY
        );
        return new Jackson2JsonMessageConverter(objectMapper);
    }

    // 如果使用Spring AMQP
    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(
            ConnectionFactory connectionFactory,
            Jackson2JsonMessageConverter jsonMessageConverter) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setMessageConverter(jsonMessageConverter);
        return factory;
    }

    @Bean
    public RabbitTemplate rabbitTemplate(CachingConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMessageConverter(jsonMessageConverter());
        rabbitTemplate.setMandatory(true);

        // 发布确认回调
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (ack) {
                log.info("消息发布确认成功: {}", correlationData);
            } else {
                log.error("消息发布确认失败: {}, 原因: {}", correlationData, cause);
                // 这里可以添加重试逻辑或告警通知
            }
        });

        // 消息无法路由回调
        rabbitTemplate.setReturnsCallback(returned -> {
            log.error("消息无法路由到队列: exchange={}, routingKey={}, replyCode={}, replyText={}, message={}",
                    returned.getExchange(),
                    returned.getRoutingKey(),
                    returned.getReplyCode(),
                    returned.getReplyText(),
                    returned.getMessage());
            // 这里可以添加重试逻辑或告警通知
        });

        return rabbitTemplate;
    }

    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(
            CachingConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setMessageConverter(jsonMessageConverter());

        // 设置消费者数量
        factory.setConcurrentConsumers(concurrency);
        factory.setMaxConcurrentConsumers(maxConcurrency);

        // 设置手动确认模式
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);

        // 设置重试策略
        RetryTemplate retryTemplate = new RetryTemplate();

        // 设置退避策略
        ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
        backOffPolicy.setInitialInterval(initialInterval);
        backOffPolicy.setMultiplier(multiplier);
        backOffPolicy.setMaxInterval(maxInterval);
        retryTemplate.setBackOffPolicy(backOffPolicy);

        // 设置重试策略
        SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
        retryPolicy.setMaxAttempts(maxAttempts);
        retryTemplate.setRetryPolicy(retryPolicy);

        factory.setRetryTemplate(retryTemplate);

        // 设置错误处理
        factory.setErrorHandler(new ConditionalRejectingErrorHandler(new FatalExceptionStrategy() {
            @Override
            public boolean isFatal(Throwable t) {
                // 判断是否是致命异常，如果是则拒绝消息并进入死信队列
                return t instanceof MessageConversionException ||
                        t instanceof MessageDeliveryException ||
                        t instanceof AmqpException;
            }
        }) {
            @Override
            public void handleError(Throwable t) {
                // 记录错误日志
                log.error("消息处理失败: {}", t.getMessage(), t);

                // 获取异常类型
                String exceptionType = t.getClass().getSimpleName();

                // 根据异常类型进行不同处理
                if (t instanceof MessageConversionException) {
                    log.error("消息转换失败，消息格式可能不正确!"+t);
                    // 可以添加告警通知
                } else if (t instanceof MessageDeliveryException) {
                    log.error("消息投递失败，可能是队列已满或消费者处理超时!"+t);
                    // 可以添加告警通知
                } else if (t instanceof AmqpException) {
                    log.error("RabbitMQ操作异常，可能是连接问题!"+t);
                    // 可以添加告警通知
                } else {
                    log.error("未知异常类型: {}!", exceptionType);
                    // 可以添加告警通知
                }

                // 调用父类方法处理异常
                super.handleError(t);
            }
        });

        // 设置消息预取数量
        factory.setPrefetchCount(10);

        // 设置消费者标签策略
        factory.setConsumerTagStrategy(queue -> "consumer-" + queue + "-" + System.currentTimeMillis());

        // 设置消费者启动超时时间
//        factory.setConsumerStartTimeout(10000);

        // 设置消费者关闭超时时间
//        factory.setConsumerStopTimeout(5000);

        return factory;
    }


}
